package com.hexinfo.dmpro.sparing.service.impl;


import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.file.FileAppender;
import cn.hutool.core.util.StrUtil;
import com.hexinfo.dmpro.common.utils.ConnExecuteUtil;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.sparing.dto.DistCPDTO;
import com.hexinfo.dmpro.sparing.dto.TableDistCPDTO;
import com.hexinfo.dmpro.sparing.model.SyncCompleteLibrary;
import com.hexinfo.dmpro.sparing.model.UatSyncLogs;
import com.hexinfo.dmpro.sparing.service.BackupSourceService;
import com.hexinfo.dmpro.sparing.service.DataRefreshService;
import com.hexinfo.dmpro.sparing.service.HotStandbyLogTableService;
import com.hexinfo.dmpro.sparing.util.Logger;
import com.xqfunds.job.core.log.XxlJobLogger;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;

@Service
@Slf4j
public class DataRefreshServiceImpl implements DataRefreshService {

    @Autowired
    @Qualifier("connPoolHiveSource")
    private DataSource connPoolHiveSource;
    @Autowired
    @Qualifier("connPoolHiveTarget")
    private DataSource connPoolHiveTarget;
    @Autowired
    @Qualifier("connPoolImpalaSource")
    private DataSource connPoolImpalaSource;
    @Autowired
    @Qualifier("connPoolImpalaTarget")
    private DataSource connPoolImpalaTarget;

    @Autowired
    private BackupSourceService backupSourceService;
    @Autowired
    private HotStandbyLogTableService hotStandbyLogTableService;

    //表完成数量
    private Map<String, Integer> tableSizeMap = new ConcurrentHashMap<String, Integer>();
    private synchronized void addTableSize(String souName) {
        tableSizeMap.put(souName, tableSizeMap.getOrDefault(souName,0)+1);
    }

    //任务结束标志
    private Map<String, Boolean> endFlagMap = new ConcurrentHashMap<String, Boolean>();
    //修改结束状态
    public synchronized void updateEndFlag(String souName,Boolean state) {
        endFlagMap.put(souName, state);
    }
    //获取结束状态
    public Boolean getEndFlag(String souName) {
        return endFlagMap.get(souName);
    }
    //删除key
    public synchronized void deleteEndFlag(String souName) {
        endFlagMap.remove(souName);
    }

    /**
     * 刷新hive、impala表
     * @return
     */
    @Override
    @Async("threadPoolTaskExecutor")
    public Future<Boolean> execShell(TableDistCPDTO tableDistCPDTO, String backupSort, int sizes, FileAppender heatlog, FileAppender tableLog) {
        souTableIsFinish(tableDistCPDTO,sizes, heatlog, tableLog);
        return new AsyncResult<>(true);
    }

    /**
     * 刷新hive、impala表
     */
    private void souTableIsFinish(TableDistCPDTO tableDistCPDTO,int sizes, FileAppender heatlog, FileAppender tableLog){
        //目标--库.表
        String targetName = tableDistCPDTO.getTargetName();
        //同步的源名
        String sourceDataName = tableDistCPDTO.getSourceDataName();
        try {
            //刷新hive目标表
            @Cleanup
            Connection connHive = connPoolHiveTarget.getConnection();
            refreshTable(connHive, targetName,heatlog,tableLog,sourceDataName,tableDistCPDTO.getTableId());
//            if (connHive != null) {
//                connHive.close();
//            }
            //刷新impala目标表
            @Cleanup
            Connection connImpala = connPoolImpalaTarget.getConnection();
            refreshTableImpala(connImpala, targetName,heatlog,tableLog,sourceDataName,tableDistCPDTO.getTableId());
//            if (connImpala != null) {
//                connImpala.close();
//            }
        }catch (Exception e){
            log.error("刷新目标表获取连接池连接异常："+e);
            logLog(heatlog,tableLog,"刷新目标表获取连接池连接异常："+e,"1");
            //表级日志分析记录修改--执行状态修改
            hotStandbyLogTableService.upState(tableDistCPDTO.getTableId(), ScanCommonConstants.ClusterName.SXSB.value);
            backupSourceService.updateTaskState(sourceDataName,false);
        }

        logLog(heatlog,tableLog,"【表刷新完成】"+targetName,"1");
        tableLog.append(logTime("-------------------------【刷新目标表结束】")+"\n"+logTime("-------------------------【数据比对开始】"));
        tableLog.flush();     //刷进日志中
        logLog(heatlog,tableLog,"","2"); //刷进日志中
        //表级日志分析记录修改--用时详情修改
        hotStandbyLogTableService.upHotLog(tableDistCPDTO.getTableId(),DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"\n");
    }

    /**
     * 刷新备份表
     * @return
     */
    private void refreshTable(Connection connection,String name,FileAppender heatlog,
                              FileAppender tableLog,String sourceDataName,String tableId) {
        String msckSql = "";
        String sqlSet = "";
        try{
            sqlSet = "set hive.msck.path.validation=ignore";
            String sql = "msck repair table {tableName}";
            msckSql = sql.replace("{tableName}", name);
            @Cleanup
            Statement st = connection.createStatement();
            //30分钟等待返回结果时长
            st.setQueryTimeout(1800);
            int iSet = st.executeUpdate(sqlSet);
            int i = st.executeUpdate(msckSql);
            tableLog.append(logTime("刷新hive的"+sqlSet+"，执行结果："+iSet+"（成功）"));
            tableLog.append(logTime("刷新hive的"+msckSql+"，执行结果："+i+"（成功）"));
        }catch (Exception e){
            e.printStackTrace();
            log.error("刷新hive备份"+sqlSet+"--"+msckSql+"异常"+e.getMessage());
            //忽略time out报错 使用第二个方案去修复分区
            if(e.getMessage().contains(ScanCommonConstants.ClusterName.SXSQLERR.value)){
                logLog(heatlog,tableLog,"刷新hive的"+sqlSet+"--"+msckSql+"异常："+e+ScanCommonConstants.ClusterName.SXTX.value,"1");
                fixPartition(name,heatlog,tableLog,sourceDataName,tableId);
            }else{
                logLog(heatlog,tableLog,"刷新hive的"+sqlSet+"--"+msckSql+"异常："+e.getMessage(),"1");
                //表级日志分析记录修改--执行状态修改
                hotStandbyLogTableService.upState(tableId, ScanCommonConstants.ClusterName.SXSB.value);
                backupSourceService.updateTaskState(sourceDataName,false);
            }
        }
//        finally {
//            try {
//                if (connection != null) {
//                    connection.close();
//                }
//            } catch (Exception e) {
//                log.error("关闭连接异常"+e);
//                logLog(heatlog,tableLog,"关闭连接异常："+e,"1");
//                //表级日志分析记录修改--执行状态修改
//                hotStandbyLogTableService.upState(tableId, ScanCommonConstants.ClusterName.SXSB.value);
//                backupSourceService.updateTaskState(sourceDataName,false);
//            }
//        }
    }

    /**
     * 刷新impala表
     * @return
     */
    private void refreshTableImpala(Connection connection,String name,FileAppender heatlog,
                                    FileAppender tableLog,String sourceDataName,String tableId) {
        String refreshSql = "";
        try{
            String sql = "invalidate metadata {tableName}";
            refreshSql = sql.replace("{tableName}", name);
            @Cleanup
            Statement st = connection.createStatement();
            //30分钟等待返回结果时长
            st.setQueryTimeout(1800);
            int i = st.executeUpdate(refreshSql);
            tableLog.append(logTime("刷新impala的"+refreshSql+"，执行结果："+i+"（成功）"));
        }catch (Exception e){
            log.error("刷新impala的"+refreshSql+"异常"+e);
            logLog(heatlog,tableLog,"刷新impala的"+refreshSql+"异常："+e,"1");
        }
//        finally {
//            try {
//                if (connection != null) {
//                    connection.close();
//                }
//            } catch (Exception e) {
//                log.error("关闭连接异常"+e);
//                logLog(heatlog,tableLog,"关闭连接异常："+e,"1");
//                //表级日志分析记录修改--执行状态修改
//                hotStandbyLogTableService.upState(tableId, ScanCommonConstants.ClusterName.SXSB.value);
//                backupSourceService.updateTaskState(sourceDataName,false);
//            }
//        }
    }

    /**
     * 按源 万国到UAT
     */
    @Override
    public String souTableIsFinishWgToUat(String targetName,UatSyncLogs uatSyncLogs){
        try {
            //刷新hive目标表
            @Cleanup
            Connection connHive = connPoolHiveTarget.getConnection();
            String hiveError = refreshTableWgToUat(connHive, targetName);
            log.warn("hive表刷新");
            if(hiveError.contains("异常")){
                if(StrUtil.isNotBlank(uatSyncLogs.getLog())){
                    uatSyncLogs.setLog(uatSyncLogs.getLog()+"\n"+"["+DateUtil.now()+"]【表刷新失败】"+hiveError);
                }
                return hiveError;
            }else{
                if(StrUtil.isNotBlank(uatSyncLogs.getLog())){
                    uatSyncLogs.setLog(uatSyncLogs.getLog()+"\n"+"["+DateUtil.now()+"]【表修复SQL】"+hiveError);
                }
                //刷新impala目标表 aresoft ty_ods ty_dwd ty_dwm 4个库 刷新impala源表
                String impalaErrorJQ = "";
                String impalaErrorWG = "";
                if(targetName.contains("aresoft.")
                        ||targetName.contains("ty_ods.")
                        ||targetName.contains("ty_dwd.")
                        ||targetName.contains("ty_dwm.")
                ){
                    @Cleanup
                    Connection connImpala = connPoolImpalaSource.getConnection();
                    impalaErrorJQ = refreshTableImpalaWgToUat(connImpala, targetName);
                    if(StrUtil.isNotBlank(uatSyncLogs.getLog())){
                        uatSyncLogs.setLog(uatSyncLogs.getLog()+"\n"+"["+DateUtil.now()+"]【Impala】"+"\n["+DateUtil.now()+"]【表刷新SQL】"+impalaErrorJQ);
                        log.warn("impala表刷新");
                    }
                }
                if(impalaErrorJQ.contains("异常")) {
                    return impalaErrorJQ;
                }else{
                    //刷新impala目标表
                    @Cleanup
                    Connection connImpala = connPoolImpalaTarget.getConnection();
                    impalaErrorWG = refreshTableImpalaWgToUat(connImpala, targetName);
                    log.warn("impala表刷新");
                    if(impalaErrorWG.contains("异常")){
                        uatSyncLogs.setLog(uatSyncLogs.getLog()+"\n"+"["+DateUtil.now()+"]【Impala】"+"\n["+DateUtil.now()+"]【uat_"+targetName+"表刷新失败】"+impalaErrorWG);
                        return impalaErrorWG;
                    }else{
                        if(StrUtil.isNotBlank(uatSyncLogs.getLog())){
                            uatSyncLogs.setLog(uatSyncLogs.getLog()+"\n"+"["+DateUtil.now()+"]【Impala】"+"\n["+DateUtil.now()+"]【表刷新成功】"+"\n["+DateUtil.now()+"]【表刷新结束】");
                        }else{
                            uatSyncLogs.setLog("["+DateUtil.now()+"]【表刷新成功】"+"\n["+DateUtil.now()+"]【表刷新结束】");
                        }
                    }
                }
            }
        }catch (Exception e){
            log.warn("表刷新异常");
            if(StrUtil.isNotBlank(uatSyncLogs.getLog())){
                uatSyncLogs.setLog(uatSyncLogs.getLog()+"\n["+DateUtil.now()+"]【"+targetName+"表刷新失败】"+e);
            }else{
                uatSyncLogs.setLog("\n["+DateUtil.now()+"]【"+targetName+"表刷新失败】"+e);
            }
            return "异常";
        }
        return "成功";
    }

    /**
     * 全库热备-表刷新
     * @param targetName
     * @return
     */
    @Override
    public String souTableIsFinishDatabase(String prefix,String targetName) {
        String cg = "";
        try {
            //刷新hive目标表
            @Cleanup
            Connection connHive = connPoolHiveTarget.getConnection();
            String hiveError = refreshTableDatabase(connHive, prefix+targetName);
            log.warn("hive表刷新");
            cg = "["+DateUtil.now()+"]【万国表修复SQL】"+hiveError;
            //刷新impala目标表 aresoft ty_ods ty_dwd ty_dwm dc aml 6个库 刷新impala源表
            String impalaErrorJQ = "";
            String impalaErrorWG = "";
//            if(targetName.contains("aresoft.")
//                    ||targetName.contains("ty_ods.")
//                    ||targetName.contains("ty_dwd.")
//                    ||targetName.contains("ty_dwm.")
//                    ||targetName.contains("dc.")
//                    ||targetName.contains("aml.")
//            ){
            if(true){
                @Cleanup
                Connection connImpala = connPoolImpalaSource.getConnection();
                impalaErrorJQ = refreshTableImpalaDatabase(connImpala, targetName);
                cg = cg+"\n["+DateUtil.now()+"]【金桥表刷新SQL】"+impalaErrorJQ;
                log.warn("金桥impala表刷新");
            }
            if(impalaErrorJQ.contains("异常")) {
                return impalaErrorJQ;
            }else{
                @Cleanup
                Connection connImpala = connPoolImpalaTarget.getConnection();
                impalaErrorWG = refreshTableImpalaDatabase(connImpala, prefix+targetName);
                cg = cg+"\n["+DateUtil.now()+"]【万国表刷新SQL】"+impalaErrorWG;
                log.warn("万国impala表刷新");
                if(impalaErrorWG.contains("异常")){
                    return impalaErrorWG;
                }else{
                    cg = cg+"\n["+DateUtil.now()+"]【表刷新成功】"+"\n["+DateUtil.now()+"]【表刷新结束】";
                }
            }
        }catch (Exception e){
            log.warn("表刷新异常");
            return "异常"+e;
        }
        return cg;
    }

    /**
     * 刷新hive表 万国到UAT
     * @return
     */
    private String refreshTableWgToUat(Connection connection,String name) {
        String msckSql = "";
        String sqlSet = "";
        try{
            sqlSet = "set hive.msck.path.validation=ignore";
            String sql = "msck repair table {tableName}";
            msckSql = sql.replace("{tableName}", name);
            @Cleanup
            Statement st = connection.createStatement();
            //30分钟等待返回结果时长
            st.setQueryTimeout(1800);
            int iSet = st.executeUpdate(sqlSet);
            int i = st.executeUpdate(msckSql);
        }catch (Exception e){
            log.error("刷新hive"+sqlSet+"--"+msckSql+"异常"+e);
            return "刷新hive"+sqlSet+"--"+msckSql+"异常"+e;
//            XxlJobLogger.log("刷新hive的"+sqlSet+"--"+msckSql+"异常："+e);
        }
//        finally {
//            try {
//                if (connection != null) {
//                    connection.close();
//                }
//            } catch (Exception e) {
//                log.error("关闭连接异常"+e);
//                return "关闭连接异常"+e;
////                XxlJobLogger.log("关闭连接异常："+e);
//            }
//        }
        return msckSql;
    }

    /**
     * 刷新hive表 全库热备
     * @return
     */
    private String refreshTableDatabase(Connection connection,String name) {
        String msckSql = "";
        String sqlSet = "";
        try{
            sqlSet = "set hive.msck.path.validation=ignore";
            String sql = "msck repair table {tableName}";
            msckSql = sql.replace("{tableName}", name);
            @Cleanup
            Statement st = connection.createStatement();
            //30分钟等待返回结果时长
            st.setQueryTimeout(1800);
            int iSet = st.executeUpdate(sqlSet);
            int i = st.executeUpdate(msckSql);
        }catch (Exception e){
            log.error("刷新hive的"+sqlSet+"--"+msckSql+"异常"+e);
            String refreshSql = "";
            try{
                @Cleanup
                Connection connectionImpala = connPoolImpalaTarget.getConnection();
                String sql = "alter table {tableName} recover partitions";
                refreshSql = sql.replace("{tableName}", name);
                @Cleanup
                Statement stImpala = connectionImpala.createStatement();
                //30分钟等待返回结果时长
                stImpala.setQueryTimeout(1800);
                int i = stImpala.executeUpdate(refreshSql);
            }catch (Exception exc){
                log.error("修复hive的"+refreshSql+"异常"+exc);
                return "修复hive的"+refreshSql+"异常"+exc+ScanCommonConstants.ClusterName.SXTX.value;
            }
//            finally {
//                try {
//                    if (connection != null) {
//                        connection.close();
//                    }
//                } catch (Exception ex) {
//                    log.error("关闭连接异常" + ex);
//                }
//            }
            return "修复hive的"+sqlSet+"--"+msckSql+"异常"+e+ScanCommonConstants.ClusterName.SXTX.value+"\n["+DateUtil.now()+"]修复hive的"+refreshSql+"成功";
        }
//        finally {
//            try {
//                if (connection != null) {
//                    connection.close();
//                }
//            } catch (Exception e) {
//                log.error("关闭连接异常"+e);
//            }
//        }
        return msckSql;
    }

    /**
     * 修复分区方案二
     * @param name
     * @param heatlog
     * @param tableLog
     * @param sourceDataName
     * @param tableId
     */
    private void fixPartition(String name,FileAppender heatlog, FileAppender tableLog,String sourceDataName,String tableId) {
        String refreshSql = "";
        try{
            @Cleanup
            Connection connection = connPoolImpalaTarget.getConnection();
            String sql = "alter table {tableName} recover partitions";
            refreshSql = sql.replace("{tableName}", name);
            @Cleanup
            Statement st = connection.createStatement();
            //30分钟等待返回结果时长
            st.setQueryTimeout(1800);
            int i = st.executeUpdate(refreshSql);
            tableLog.append(logTime("修复hive的"+refreshSql+"，执行结果："+i+"（成功）"));
        }catch (Exception e){
            log.error("修复hive的"+refreshSql+"异常"+e);
            logLog(heatlog,tableLog,"修复hive的"+refreshSql+"异常："+e+ScanCommonConstants.ClusterName.SXTX.value,"1");
        }
//        finally {
//            try {
//                if (connection != null) {
//                    connection.close();
//                }
//            } catch (Exception e) {
//                log.error("关闭连接异常"+e);
//                logLog(heatlog,tableLog,"关闭连接异常："+e,"1");
//                //表级日志分析记录修改--执行状态修改
//                hotStandbyLogTableService.upState(tableId, ScanCommonConstants.ClusterName.SXSB.value);
//                backupSourceService.updateTaskState(sourceDataName,false);
//            }
//        }
    }

    /**
     * 刷新impala表 全库热备
     * @return
     */
    private String refreshTableImpalaDatabase(Connection connection,String name) {
        String refreshSql = "";
        try{
            String sql = "invalidate metadata {tableName}";
            refreshSql = sql.replace("{tableName}", name);
            @Cleanup
            Statement st = connection.createStatement();
            //30分钟等待返回结果时长
            st.setQueryTimeout(1800);
            int i = st.executeUpdate(refreshSql);
        }catch (Exception e){
            log.error("刷新impala的"+refreshSql+"异常"+e);
            return "刷新impala的"+refreshSql+"异常"+e;
        }
//        finally {
//            try {
//                if (connection != null) {
//                    connection.close();
//                }
//            } catch (Exception e) {
//                log.error("关闭连接异常"+e);
//                return "关闭连接异常："+e;
//            }
//        }
        return refreshSql;
    }


    /**
     * 刷新impala表 万国到UAT
     * @return
     */
    private String refreshTableImpalaWgToUat(Connection connection,String name) {
        String refreshSql = "";
        try{
            String sql = "invalidate metadata {tableName}";
            refreshSql = sql.replace("{tableName}", name);
            @Cleanup
            Statement st = connection.createStatement();
            //30分钟等待返回结果时长
            st.setQueryTimeout(1800);
            int i = st.executeUpdate(refreshSql);
        }catch (Exception e){
            log.error("刷新impala的"+refreshSql+"异常"+e);
            return "刷新impala的"+refreshSql+"异常"+e;
//            XxlJobLogger.log("刷新impala的"+refreshSql+"异常："+e);
        }
//        finally {
//            try {
//                if (connection != null) {
//                    connection.close();
//                }
//            } catch (Exception e) {
//                log.error("关闭连接异常"+e);
//                return "关闭连接异常"+e;
////                XxlJobLogger.log("关闭连接异常："+e);
//            }
//        }
        return refreshSql;
    }

    /**
     * 加锁，防止并发问题
     * @param heatlog
     * @param tableLog
     * @param logStr
     * @param str
     */
    private synchronized void logLog(FileAppender heatlog, FileAppender tableLog, String logStr ,String str) {
        if (str.equals("1")){
            heatlog.append(logTime(logStr));
            tableLog.append(logTime(logStr));
        }else {
            heatlog.flush();     //刷进日志中
        }
    }

    /**
     * 日志加日期时间
     * @param str
     * @return
     */
    private String logTime(String str) {
        return "["+DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"]" +str;
    }

}
